/*
  * Copyright (c) 2012-2014 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package org.dna.mqtt.moquette.processor.impl;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
import org.dna.mqtt.moquette.messaging.spi.impl.DebugUtils;
import org.dna.mqtt.moquette.messaging.spi.impl.FileAuthenticator;
import org.dna.mqtt.moquette.messaging.spi.impl.HawtDBPersistentStore;
import org.dna.mqtt.moquette.messaging.spi.impl.MessageOutputDisruptor;
import org.dna.mqtt.moquette.messaging.spi.impl.ValueEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.InitEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.LostConnectionEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.OutputMessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.StopEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
import org.dna.mqtt.moquette.processor.ProtocolProcessor;
import org.dna.mqtt.moquette.processor.ProtocolProcessorFactory;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.IAuthenticator;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lmax.disruptor.EventHandler;

/**
 * Class responsible to handle the logic of MQTT protocol it's the director of
 * the protocol execution. 
 * 
 * @author andrea
 */
public class MqttProtocolProcessor implements EventHandler<ValueEvent> {
	
	public static final class WillMessage {
        private final String topic;
        private final ByteBuffer payload;
        private final boolean retained;
        private final QOSType qos;

        public WillMessage(String topic, ByteBuffer payload, boolean retained, QOSType qos) {
            this.topic = topic;
            this.payload = payload;
            this.retained = retained;
            this.qos = qos;
        }

        public String getTopic() {
            return topic;
        }

        public ByteBuffer getPayload() {
            return payload;
        }

        public boolean isRetained() {
            return retained;
        }

        public QOSType getQos() {
            return qos;
        }   
    }
    private static final Logger LOG = LoggerFactory.getLogger(MqttProtocolProcessor.class);
    
    private MessageOutputDisruptor m_outputDisruptor;
    
    /* Maps clientIDs to connection descriptor */
    private Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
    
    /* Maps clientID to Will testament, if specified on CONNECT */
    private Map<String, WillMessage> m_willStore = new HashMap<String, WillMessage>();
    
    /* Factory class that creates processor for each protocol command */
    private ProtocolProcessorFactory m_protocolProcessorFactory = new MqttProtocolProcessorFactory();
    
    /* Network channel or session */
    private ServerChannel m_session;
    
    /* The authenticator used in connect messages */
    private IAuthenticator m_authenticator;
    
    /* The subscription store where are stored all the existing clients subscriptions */
    private SubscriptionsStore m_subscriptionStore;
    
    /* The persistent store to use for save/load of messages for QoS1 and QoS2 handling*/
    private IMessagesStore m_messagesStore;
    
    /* The clients sessions store, used to persist subscriptions */
    private ISessionsStore m_sessionsStore;
    
    /* Base protocol processor interface implemented by all protocol command processors */
    private ProtocolProcessor m_protocolProcessor;
    
    private CountDownLatch m_stopCountDownLatch;
    
    public MqttProtocolProcessor() {
    	//Empty
    }
    
    public MqttProtocolProcessor(CountDownLatch stopCountDownLatch) {
    	this.m_stopCountDownLatch = stopCountDownLatch;
    }
    
    /**
     * Init method, used mostly for testing
	 * @param subscriptions the subscriptions tore
	 * @param m_storageService the storage store
	 * @param m_sessionStore the session store
	 * @param m_mockAuthenticator the authenticator
	 */
	public void init(SubscriptionsStore subscriptionsStore,
			IMessagesStore storageService, ISessionsStore sessionStore, IAuthenticator authenticator) {
		this.m_subscriptionStore = subscriptionsStore;
		this.m_messagesStore = storageService;
		this.m_sessionsStore = sessionStore;
		this.m_authenticator = authenticator;
	}
	
    /**
     * Invoked with LMAX disruptor event see {@cod EventHandler}
     */
    public void onEvent(ValueEvent valueEvent, long l, boolean bln)
			throws Exception {
		LOG.info("onEvent processing messaging event from input ringbuffer {}", valueEvent);
		
		MessagingEvent evt = valueEvent.getEvent();
		if (evt instanceof InitEvent) {
			processInit(((InitEvent) evt).getConfig());
		} else if (evt instanceof StopEvent) {
			processStop();
		} else if (evt instanceof LostConnectionEvent) {
			LostConnectionEvent lostEvt = (LostConnectionEvent) evt;
			proccessConnectionLost(lostEvt.getClientID());
		} else {
			processProtocolEvent(valueEvent);
		}
	}
    
    public boolean clientExists(String clientID) {
    	return m_clientIDs.containsKey(clientID);
    }
    
    public ServerChannel getClientSession(String clientID) {
    	return m_clientIDs.get(clientID).getSession();
    }
    
    public void addClient(String clientID, ConnectionDescriptor connectionDescriptor) {
    	m_clientIDs.put(clientID, connectionDescriptor);
    }
    
    public void addWillMessage(String clientID, WillMessage willMessage) {
    	m_willStore.put(clientID, willMessage);
    }
    
    public boolean checkIfClientValid(String userName, String password) {
    	return m_authenticator.checkValid(userName, password);
    }
    
    public void activeClient(String clientID) {
    	m_subscriptionStore.activate(clientID);
    }
    
    public void cleanSession(String clientID) {
        LOG.info("cleaning old saved subscriptions for client <{}>", clientID);

        //remove from log all subscriptions
        m_sessionsStore.wipeSubscriptions(clientID);
        m_subscriptionStore.removeForClient(clientID);

        //remove also the messages stored of type QoS1/2
        m_messagesStore.cleanPersistedPublishes(clientID);
    }
    
    public void republishStored(String clientID) {
        LOG.trace("republishStored invoked");
        List<PublishEvent> publishedEvents = m_messagesStore.retrievePersistedPublishes(clientID);
        if (publishedEvents == null) {
            LOG.info("No stored messages for client <{}>", clientID);
            return;
        }

        LOG.info("republishing stored messages to client <{}>", clientID);
        for (PublishEvent pubEvt : publishedEvents) {
            sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
                   pubEvt.getMessage(), false, pubEvt.getMessageID());
        }
    }
    
    public void addSubscription(Subscription subscription, String clientID) {
    	 m_sessionsStore.addNewSubscription(subscription, clientID);
         m_subscriptionStore.add(subscription);
    }
    
    public void removeSubscription(String topic, String clientID) {
    	m_subscriptionStore.removeSubscription(topic, clientID);
    }
    
    public Collection<HawtDBPersistentStore.StoredMessage> getRetainedMessagesForTopic(String topic) {
    	final String topicName = topic;
    	Collection<HawtDBPersistentStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
            public boolean match(String key) {
                return  SubscriptionsStore.matchTopics(key, topicName);
            }
        });
    	return messages;
    }
    
    public PublishEvent retrieveQoS2Message(String publishKey) {
    	return m_messagesStore.retrieveQoS2Message(publishKey);
    }
    
    public void removeQoS2Message(String publishKey) {
    	m_messagesStore.removeQoS2Message(publishKey);
    }
    
    public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) {
    	m_messagesStore.storeRetained(topic, message, qos);
    }
    
    public void cleanInFlight(String publishKey) {
    	m_messagesStore.cleanInFlight(publishKey);
    }
    
    public void cleanPersistedPublishMessage(String clientID, Integer messageID) {
		m_messagesStore.cleanPersistedPublishMessage(clientID, messageID);
    }

    private void processInit(Properties props) {
    	LOG.debug("processInit invoked");
    	LOG.debug("Initializing different stores...");
    	
    	m_outputDisruptor = MessageOutputDisruptor.getInstance();
    	
    	HawtDBPersistentStore hawtStorage = new HawtDBPersistentStore();
		this.m_messagesStore = hawtStorage;
		this.m_messagesStore.initStore();
		
		this.m_sessionsStore = hawtStorage;

		List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
		this.m_subscriptionStore = new SubscriptionsStore();
		this.m_subscriptionStore.init(storedSubscriptions);
		
		//Initialize authenticator
		String passwdPath = props.getProperty("password_file");
		String configPath = System.getProperty("moquette.path", null);
		this.m_authenticator = new FileAuthenticator(configPath, passwdPath);
		
		LOG.debug("MqttProtocolProcessor initialization completed successfully");
	}
    
    private void processStop() {
		LOG.debug("processStop invoked");
		
		m_messagesStore.close();
		m_subscriptionStore = null;
		
		m_stopCountDownLatch.countDown();
	}
    
    private void proccessConnectionLost(String clientID) {
    	LOG.debug("proccessConnectionLost invoked");
    	//If already removed a disconnect message was already processed for this clientID
        if (m_clientIDs.remove(clientID) != null) {
            //de-activate the subscriptions for this ClientID
            m_subscriptionStore.deactivate(clientID);
            LOG.info("Lost connection with client <{}>", clientID);
        }
        //TODO: publish the Will message (if any) for the clientID
        if (m_willStore.containsKey(clientID)) {
            WillMessage will = m_willStore.get(clientID);
            PublishEvent pubEvt = new PublishEvent(will.getTopic(), will.getQos(), will.getPayload(), will.isRetained(), clientID);
            
            processPublish(pubEvt);
            m_willStore.remove(clientID);
        }
        LOG.debug("proccessConnectionLost completed successfully");
    }
    
    public void processProtocolEvent(ValueEvent valueEvent) {
    	LOG.debug("processProtocolEvent invoked");
    	
		m_protocolProcessor = m_protocolProcessorFactory.getProcessor(valueEvent, this);
		if (m_protocolProcessor != null) {
			m_protocolProcessor.processEvent(valueEvent);
		} else {
			LOG.error("Illegal message received:"+ valueEvent.toString());
			throw new RuntimeException("Illegal message received " + valueEvent.toString());
		}
	}

    public void processDisconnect(String clientID, boolean cleanSession) throws InterruptedException {
        if (cleanSession) {
        	//cleanup topic subscriptions
            cleanSession(clientID);
        }
        m_clientIDs.remove(clientID);
        m_session.close(true);

        //de-activate the subscriptions for this ClientID
        m_subscriptionStore.deactivate(clientID);
        
        //cleanup the will store
        m_willStore.remove(clientID);
        LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession);
    }

    public void processPublish(PublishEvent evt) {
        LOG.trace("PUB --PUBLISH--> SRV processPublish invoked with {}", evt);
        final String topic = evt.getTopic();
        final AbstractMessage.QOSType qos = evt.getQos();
        final ByteBuffer message = evt.getMessage();
        boolean retain = evt.isRetain();
        
        LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}", 
                evt.getClientID(), evt.getTopic(), evt.getQos());

        String publishKey = null;
        if (qos == AbstractMessage.QOSType.LEAST_ONE) {
            //store the temporary message
            publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID());
            m_messagesStore.addInFlight(evt, publishKey);
        }  else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
            publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID());
            //store the message in temp store
            m_messagesStore.persistQoS2Message(publishKey, evt);
            sendPubRec(evt.getClientID(), evt.getMessageID());
        }

        //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
        if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) {
            publish2Subscribers(topic, qos, message, retain, evt.getMessageID());
        }

        if (qos == AbstractMessage.QOSType.LEAST_ONE) {
            if (publishKey == null) {
                throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + evt);
            }
            m_messagesStore.cleanInFlight(publishKey);
            sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
            LOG.debug("replying with PubAck to MSG ID {}", evt.getMessageID());
        }

        if (retain) {
            m_messagesStore.storeRetained(topic, message, qos);
        }
    }
    
    /**
     * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
     */
    public void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer origMessage, boolean retain, Integer messageID) {
        LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
        if (LOG.isDebugEnabled()) {
            LOG.debug("content <{}>", DebugUtils.payload2Str(origMessage));
            LOG.debug("subscription tree {}", m_subscriptionStore.dumpTree());
        }
        for (final Subscription sub : m_subscriptionStore.matches(topic)) {
            if (qos.ordinal() > sub.getRequestedQos().ordinal()) {
                qos = sub.getRequestedQos();
            }
            
            ByteBuffer message = origMessage.duplicate();
            LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", 
                    sub.getClientId(), sub.getTopicFilter(), qos, sub.isActive());
            
            if (qos == AbstractMessage.QOSType.MOST_ONE && sub.isActive()) {
                //QoS 0
                sendPublish(sub.getClientId(), topic, qos, message, false);
            } else {
                //QoS 1 or 2
                //if the target subscription is not clean session and is not connected => store it
                if (!sub.isCleanSession() && !sub.isActive()) {
                    //clone the event with matching clientID
                    PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID);
                    m_messagesStore.storePublishForFuture(newPublishEvt);
                } else  {
                    //if QoS 2 then store it in temp memory
                    if (qos ==AbstractMessage.QOSType.EXACTLY_ONCE) {
                        String publishKey = String.format("%s%d", sub.getClientId(), messageID);
                        PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID);
                        m_messagesStore.addInFlight(newPublishEvt, publishKey);
                    }
                    //publish
                    if (sub.isActive()) {
                        sendPublish(sub.getClientId(), topic, qos, message, false);
                    }
                }
            }
        }
    }
    
    
    public void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
        //TODO pay attention to the message ID can't be 0 and it's the message sent to subscriber
        int messageID = 1;
        sendPublish(clientId, topic, qos, message, retained, messageID);
    }
   
    public void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) {
        LOG.debug("sendPublish invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID);
        PublishMessage pubMessage = new PublishMessage();
        pubMessage.setRetainFlag(retained);
        pubMessage.setTopicName(topic);
        pubMessage.setQos(qos);
        pubMessage.setPayload(message);
        
        LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
        if (LOG.isDebugEnabled()) {
            LOG.debug("content <{}>", DebugUtils.payload2Str(message));
        }
        if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
            pubMessage.setMessageID(messageID);
        }

        if (m_clientIDs == null) {
            throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
        }
        LOG.debug("clientIDs are {}", m_clientIDs);
        if (m_clientIDs.get(clientId) == null) {
            throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs));
        }
        LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession());
        disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage));
    }
       
    private void sendPubRec(String clientID, int messageID) {
        LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
        PubRecMessage pubRecMessage = new PubRecMessage();
        pubRecMessage.setMessageID(messageID);

        disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRecMessage));
    }
    
    private void sendPubAck(PubAckEvent evt) {
        LOG.trace("sendPubAck invoked");

        String clientId = evt.getClientID();

        PubAckMessage pubAckMessage = new PubAckMessage();
        pubAckMessage.setMessageID(evt.getMessageId());

        try {
            if (m_clientIDs == null) {
                throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
            }
            LOG.debug("clientIDs are {}", m_clientIDs);
            if (m_clientIDs.get(clientId) == null) {
                throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs));
            }
            disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubAckMessage));
        }catch(Throwable t) {
            LOG.error(null, t);
        }
    }
    
    public void disruptorPublish(OutputMessagingEvent msgEvent) {
    	m_outputDisruptor.disruptorPublish(msgEvent);
    }
}
